// Copyright 2013 The Gorilla WebSocket Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package main

import (
	"bytes"
	"fmt"
	"net"
	"net/http"
	"os"
	"strings"

	"github.com/gorilla/websocket"

	"encoding/json"
	"time"
	log "ycx-glog"

	"chat/common"
	"chat/db"
)

const (
	// Time allowed to write a message to the peer.
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the peer.
	pongWait = 60 * time.Second

	// Send pings to peer with this period. Must be less than pongWait.
	pingPeriod = (pongWait * 9) / 10

	// Maximum message size allowed from peer.
	maxMessageSize = 512

	MAX_READ_RECORDS = 500 //每次从数据库读取记录数
	MAX_ORDER_CHANS  = 500 //队列最大可缓存数
	MAX_THREADS      = 100 //任务同时执行线程数
)

type doParam struct {
	Addr string
	Msg  string
	Time string
}

var orders = make(chan doParam, MAX_ORDER_CHANS)

var (
	newline = []byte{'\n'}
	space   = []byte{' '}
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
}

// Client is a middleman between the websocket connection and the hub.
type Client struct {
	hub *Hub

	// The websocket connection.
	conn *websocket.Conn

	// Buffered channel of outbound messages.
	send chan []byte
}

// readPump pumps messages from the websocket connection to the hub.
//
// The application runs readPump in a per-connection goroutine. The application
// ensures that there is at most one reader on a connection by executing all
// reads from this goroutine.
func (c *Client) readPump() {
	defer func() {
		c.hub.unregister <- c
		c.conn.Close()
	}()
	c.conn.SetReadLimit(maxMessageSize)
	c.conn.SetReadDeadline(time.Now().Add(pongWait))
	c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
	for {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Infoln("error: %v", err)
			}
			break
		}
		message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))
		fmt.Println(string(message))
		c.hub.broadcast <- message

		// 接收消息放入通道 同时应该加入到数据库
		msgSql := "INSERT INTO msg (addr,userid,msg,time,readstate,sendstate) VALUES (?,?,?,?,?,?)"
		_, err = db.SDB().Exec(msgSql,
			"192.168.3.110:8366",
			"999",
			string(message),
			common.TimeFormat(time.Now(), "yyyy-MM-dd HH:mm:ss"),
			"SUCCESS",
			"wait")
		if err != nil {
			log.Infoln("msg insert err:", err)
		}
	}
}

// writePump pumps messages from the hub to the websocket connection.
//
// A goroutine running writePump is started for each connection. The
// application ensures that there is at most one writer to a connection by
// executing all writes from this goroutine.
func (c *Client) writePump() {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		c.conn.Close()
	}()
	for {
		select {
		case message, ok := <-c.send:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				// The hub closed the channel.
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}

			w, err := c.conn.NextWriter(websocket.TextMessage)
			if err != nil {
				return
			}
			w.Write(message)

			// Add queued chat messages to the current websocket message.
			n := len(c.send)
			for i := 0; i < n; i++ {
				w.Write(newline)
				w.Write(<-c.send)
			}

			if err := w.Close(); err != nil {
				return
			}
		case <-ticker.C:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

// serveWs handles websocket requests from the peer.
func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request, udpconn *net.UDPConn) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Infoln(err)
		return
	}
	client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)}
	client.hub.register <- client

	// Allow collection of memory referenced by the caller by doing all work in
	// new goroutines.
	go client.writePump()
	go client.readPump()

	// 转发新消息
	go client.writeUdp(udpconn)
	// 接收消息
	go client.readUdp(udpconn)
}

func (c *Client) readUdp(conn *net.UDPConn) {
	// ①接收数据
	var buf [512]byte
	n, raddr, err := conn.ReadFromUDP(buf[0:])
	if err != nil {
		return
	}
	log.Infoln("raddr is ", raddr)
	req := make(map[string]interface{})
	json.Unmarshal(buf[0:n], &req)
	log.Info(req)
	log.Infoln("msg is ", string(buf[0:n]))
	log.Infoln("byte msg is ", []byte(common.ValueToString(req["msg"], "")))

	c.hub.broadcast <- []byte(common.ValueToString(req["msg"], ""))

	// ②raddr + userid + msgid + msg + time 绑定在数据库
	msgSql := "INSERT INTO msg (addr,userid,msg,time,readstate,sendstate) VALUES (?,?,?,?,?,?)"
	_, err = db.SDB().Exec(msgSql,
		raddr.String(),
		req["userid"],
		req["msg"],
		common.TimeFormat(time.Now(), "yyyy-MM-dd HH:mm:ss"),
		"SUCCESS",
		"wait")
	if err != nil {
		log.Infoln("msg insert err:", err)
	}

	log.Infoln("insert msg")
	sql := "SELECT * FROM user WHERE ip=?"
	rows, err := db.SDB().Queryx(sql, raddr.String())
	if err != nil {
		return
	}
	rs := common.Rows(rows)
	rows.Close()
	log.Infoln("SELECT user")

	if len(rs) == 0 {
		userSql := "INSERT INTO user (nickname,userid,ip,groupid) VALUES (?,?,?,?)"
		db.SDB().Exec(userSql,
			req["nickname"],
			req["userid"],
			raddr.String(),
			1)
		log.Infoln("insert user")
	}
}

func (c *Client) writeUdp(conn *net.UDPConn) {
	go doing(conn)

	// 查询未处理的消息
	for {
		seleSql := "SELECT * FROM msg WHERE sendstate = ?"
		rows, err := db.SDB().Queryx(seleSql, "wait")
		if err != nil {
			return
		}
		list := common.Rows(rows)
		rows.Close()

		ids := ""
		if len(list) > 0 {
			for _, v := range list {
				ids = ids + common.ValueToString(v["id"], "") + ","
				ids = strings.TrimRight(ids, ",")
				usql := "UPDATE msg SET sendstate = ? WHERE id IN (" + ids + ")"
				_, err = db.SDB().Exec(usql, "close")
				if err != nil {
					log.Errorln("update msg close", err)
				}

				orders <- doParam{
					Addr: common.ValueToString(v["addr"], ""),
					Msg:  common.ValueToString(v["msg"], ""),
					Time: common.ValueToTime(v["time"], "", "0000-00-00 00:00:00"),
				}
			}
		}
	}
}

//从chan队列及时取出记录, 分发消息
func doing(conn *net.UDPConn) {
	lrun := common.NewLimitRun(MAX_THREADS)

	for {
		select {
		case task, ok := <-orders:
			if !ok {
				log.Info("消息发送终止")
				return //停机退出
			}

			//cond 状态
			ctms := time.Now()
			lrun.Wait()
			ctme := time.Now()
			log.Infoln("%0.3fs %s", ctme.Sub(ctms).Seconds(), lrun.String())

			go _doing(task, lrun, conn)
		}
	}
}

func _doing(task doParam, lrun *common.LimitRun, conn *net.UDPConn) {
	defer lrun.Signal()

	seleSql := "SELECT * FROM user WHERE ip != ?"
	rows, err := db.SDB().Queryx(seleSql, task.Addr)
	if err != nil {
		return
	}
	list := common.Rows(rows)
	rows.Close()

	for _, v := range list {
		list := strings.Split(common.ValueToString(v["ip"], ""), ":")
		ip := net.ParseIP(list[0])
		dstAddr := &net.UDPAddr{IP: ip, Port: common.ValueToInt(list[1], 0)}

		_, err := conn.WriteToUDP([]byte(task.Msg), dstAddr)
		if err != nil {
			return
		}
	}
}

//func serveUdp(hub *Hub, w http.ResponseWriter, r *http.Request, udpconn *net.UDPConn) {
//	conn, err := upgrader.Upgrade(w, r, nil)
//	if err != nil {
//		log.Infoln(err)
//		return
//	}
//	client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)}
//	client.hub.register <- client

//}

func checkError(err error) {
	if err != nil {
		log.Infoln("Error: %s", err.Error())
		os.Exit(1)
	}
}
